package com.education.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.app.BaseAppV1;
import com.education.realtime.bean.TradeOrderDetailBean;
import com.education.realtime.common.Constant;
import com.education.realtime.fuction.DimAsyncFunction;
import com.education.realtime.util.AtguiguUtil;
import com.education.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/* 下单情况统计
    当日所有学科 的 下单次数、下单人数、下单金额
    当日所有类别的 下单次数、下单人数、下单金额
    当日所有课程的下单人数、下单金额
    一行数据表示一个用户对一个学科的一次下单记录
*/
/*
    一个课程中包含多个学科，一个学科属于一个类别
 */
public class DwsTradeOrderDetailWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsTradeOrderDetailWindow().init(4014,"DwsTradeOrderDetailWindow", "dwd_trade_order_detail_app");
    }


    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1.转换数据为pojo
        SingleOutputStreamOperator<TradeOrderDetailBean> beanStream = beanStream(stream);
        // 2.丰富维度
        SingleOutputStreamOperator<TradeOrderDetailBean> dimStream = asyncStream(beanStream);

        // 2.封装包装类
        SingleOutputStreamOperator<TradeOrderDetailBean> subjectStream = beanSubjectStream(dimStream);
        // 3.封装课程包装类
        SingleOutputStreamOperator<TradeOrderDetailBean> courseStream = beanCourseStream(dimStream);

        // 4.全局开窗聚合得到所有课程的聚合
        SingleOutputStreamOperator<TradeOrderDetailBean> reduceStream = windowAndReduce(subjectStream,courseStream);
        // 4.写入clickhouse
        reduceStream.addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB, "dws_trade_course_subject_category_order_detail", TradeOrderDetailBean.class));

    }




    //所有数据开窗聚合
    private SingleOutputStreamOperator<TradeOrderDetailBean> windowAndReduce(SingleOutputStreamOperator<TradeOrderDetailBean> beanStream, SingleOutputStreamOperator<TradeOrderDetailBean> courseStream) {
        return beanStream.union(courseStream)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<TradeOrderDetailBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((obj,e)->obj.getTs()))
                .keyBy(ods->ods.getCourseId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeOrderDetailBean>() {
                            @Override
                            public TradeOrderDetailBean reduce(TradeOrderDetailBean v1, TradeOrderDetailBean v2) throws Exception {
                                v1.setCourseOrderUser(v1.getCourseOrderUser() + v2.getCourseOrderUser());
                                v1.setCourseOrderAmount(v1.getCourseOrderAmount() + v2.getCourseOrderAmount());
                                v1.setSubjectOrderCount(v1.getSubjectOrderCount() + v2.getSubjectOrderCount());
                                v1.setSubjectOrderUser(v1.getSubjectOrderUser() + v2.getSubjectOrderUser());
                                v1.setSubjectOrderAmount(v1.getSubjectOrderAmount() + v2.getSubjectOrderAmount());
                                v1.setCategoryOrderCount(v1.getCategoryOrderCount() + v2.getCategoryOrderCount());
                                v1.setCategoryOrderUser(v1.getCategoryOrderUser() + v2.getCategoryOrderUser());
                                v1.setCategoryOrderAmount(v1.getCategoryOrderAmount() + v2.getCategoryOrderAmount());
                                return v1;
                            }
                        }
                        , new ProcessWindowFunction<TradeOrderDetailBean, TradeOrderDetailBean, String, TimeWindow>() {
                            @Override
                            public void process(String key, Context ctx, Iterable<TradeOrderDetailBean> iterable, Collector<TradeOrderDetailBean> out) throws Exception {
                                TradeOrderDetailBean next = iterable.iterator().next();
                                next.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                next.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                out.collect(next);
                            }
                        });
    }

    // 封装课程的包装类
    private SingleOutputStreamOperator<TradeOrderDetailBean> beanCourseStream(SingleOutputStreamOperator<TradeOrderDetailBean> stream) {
        return stream
                .keyBy(obj -> obj.getUserId() + "_"  + obj.getCourseId())
                .process(new KeyedProcessFunction<String, TradeOrderDetailBean, TradeOrderDetailBean>() {

                    private ValueState<String> lateDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lateDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lateDate", String.class));
                    }

                    @Override
                    public void processElement(TradeOrderDetailBean obj, Context ctx, Collector<TradeOrderDetailBean> out) throws Exception {
                        String value = lateDate.value();
                        String today = AtguiguUtil.toDate(obj.getTs());
                        if (!today.equals(value)) {
                            out.collect(TradeOrderDetailBean.builder()
                                    .courseOrderUser(1L)
                                    .build()
                            );
                        }
                    }
                })
                ;
    }

    // 封装包装类，得到一个用户对一个学科的bena对象
    private SingleOutputStreamOperator<TradeOrderDetailBean> beanSubjectStream(SingleOutputStreamOperator<TradeOrderDetailBean> stream) {
        return stream
                .keyBy(obj -> obj.getUserId() + "_"  + obj.getSubjectId())
                .process(new KeyedProcessFunction<String, TradeOrderDetailBean, TradeOrderDetailBean>() {

                    private ValueState<String> lateDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lateDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lateDate", String.class));
                    }

                    @Override
                    public void processElement(TradeOrderDetailBean obj, Context ctx, Collector<TradeOrderDetailBean> out) throws Exception {
                        String value = lateDate.value();
                        String today = AtguiguUtil.toDate(obj.getTs());
                        if (!today.equals(value)) {
                            out.collect(TradeOrderDetailBean.builder()
                                    .subjectOrderUser(1L)
                                    .subjectOrderCount(1L)
                                    .categoryOrderCount(1L)
                                    .categoryOrderUser(1L)
                                    .build()
                            );
                        } else {
                            out.collect(TradeOrderDetailBean.builder()
                                    .subjectOrderCount(1L)
                                    .categoryOrderCount(1L)
                                    .build());
                        }
                    }
                })
                ;
    }

    // 2.丰富维度
    private SingleOutputStreamOperator<TradeOrderDetailBean> asyncStream(SingleOutputStreamOperator<TradeOrderDetailBean> stream) {
        SingleOutputStreamOperator<TradeOrderDetailBean> subjectStream = AsyncDataStream.unorderedWait(
                stream,
                new DimAsyncFunction<TradeOrderDetailBean>() {
                    @Override
                    public void addDim(TradeOrderDetailBean input, JSONObject dim) {
                        input.setSubjectId(dim.getString("subject_id"));
                        input.setSubjectOrderAmount(dim.getDoubleValue("actual_price"));
                        input.setCategoryOrderAmount(dim.getDoubleValue("actual_price"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getDimId(TradeOrderDetailBean input) {
                        return input.getCourseId();
                    }
                },
                120,
                TimeUnit.SECONDS
        );
        return AsyncDataStream.unorderedWait(
                subjectStream,
                new DimAsyncFunction<TradeOrderDetailBean>() {
                    @Override
                    public void addDim(TradeOrderDetailBean input, JSONObject dim) {
                        input.setCategoryId(dim.getString("category_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_subject_info";
                    }

                    @Override
                    public String getDimId(TradeOrderDetailBean input) {
                        return input.getSubjectId();
                    }
                },
                120,
                TimeUnit.SECONDS
        );
    }

    // 1.转换为bean
    private SingleOutputStreamOperator<TradeOrderDetailBean> beanStream(DataStreamSource<String> stream) {
        return stream.map(new MapFunction<String, TradeOrderDetailBean>() {
            @Override
            public TradeOrderDetailBean map(String value) throws Exception {
                JSONObject obj = JSON.parseObject(value);
                return TradeOrderDetailBean.builder()
                        .userId(obj.getString("user_id"))
                        .courseId(obj.getString("course_id"))
                        .courseOrderAmount(obj.getDoubleValue("final_amount"))
                        .ts(obj.getLong("ts")*1000)
                        .build();

            }
        });
    }

}
